package com.diver.flinkdemo.sink;

import com.diver.flinkdemo.entity.DevUploadData;
import com.diver.flinkdemo.mapper.dorisMapper.DevUploadDataMapper;
import com.diver.flinkdemo.utils.DynamicDataSourceUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;

import java.util.Date;

/**
 * @author LuJW
 * @date 2023/3/29 10:01
 */
@Slf4j
public class DevUploadDataSink extends RichSinkFunction<DevUploadData> {
    /**
     * sqlSessionFactory
     */
    private SqlSessionFactory sqlSessionFactory;


    @Override
    public void open(Configuration parameters) throws Exception {
//        log.info("DataSourceFactoryUtil->{}",DataSourceFactoryUtil);
        if (DynamicDataSourceUtil.dorisSessionFactory == null) {
            DynamicDataSourceUtil sourceUtil = new DynamicDataSourceUtil();
            sourceUtil.init();
            log.info("二次初始化完成...");
        }
        log.info("初始化获取sqlSessionFactory->{}", DynamicDataSourceUtil.dorisSessionFactory);
        sqlSessionFactory = DynamicDataSourceUtil.dorisSessionFactory;

    }

    @Override
    public void invoke(DevUploadData value, Context context) throws Exception {
        SqlSession sqlSession = null;
        try {
            //获取sqlSession
            log.info("sqlSessionFactory->{}",sqlSessionFactory);
            sqlSession = sqlSessionFactory.openSession();
            value.setCreateTime(new Date());
            sqlSession.getMapper(DevUploadDataMapper.class).save(value);
            log.info("save complete!");
        } catch (Exception e) {
            log.error(this.getClass().getName() + ".invoke()执行失败", e);
            //重试一次
            try {
                if (sqlSession != null) {
//                    sqlSession.getMapper(TestMapper.class).batchSave(values);
                    value.setCreateTime(new Date());
                    sqlSession.getMapper(DevUploadDataMapper.class).insert(value);
                }
            } catch (Exception ex) {
                log.error(this.getClass().getName() + ".invoke()重试执行失败", e);
            }
        } finally {
            //关闭sqlSession
            if (sqlSession != null) {
                log.info("session commit -> session close");
                DynamicDataSourceUtil.sessionDoAfter(sqlSession);

            }
        }
    }
}
